package kafka

import (
    "fmt"
    "github.com/Shopify/sarama"
    "log"
    "os"
    "os/signal"
    "time"
)

type LogData struct {
    data  string
    topic string
}
var (
    client sarama.SyncProducer
    logDataChan chan *LogData
)

func InitKaka(address []string,maxSize int) (err error){
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    config.Producer.Return.Successes = true
    // 连接kafka
    client, err = sarama.NewSyncProducer(address, config)
    if err != nil {
        fmt.Println("producer closed, err:", err)
        return
    }
    logDataChan=make(chan *LogData,maxSize)

    go SendMessageToKafka()
    return
}

func SendToLogDataChan(topic string ,data string)  {
    logData:=&LogData{
        data: data,
        topic: topic,
    }
    logDataChan <- logData

}

func SendMessageToKafka() {

    msg := new(sarama.ProducerMessage)



    for  {
        select {
        case logdata:=<- logDataChan:
            msg.Topic = logdata.topic
            msg.Value = sarama.StringEncoder(logdata.data)
            pid, offset, err := client.SendMessage(msg)
            if err != nil {
                fmt.Printf("send msg failed, err: %v\n", err)
                return
            }
            fmt.Printf("pid:%v offset:%v\n", pid, offset)
        default:
            time.Sleep(time.Millisecond*50)
        }
    }


}
func ConsumerMessage(topic string,address []string){

    consumer, err := sarama.NewConsumer(address, nil)
    if err != nil {
        panic(err)
    }

    defer func() {
        if err := consumer.Close(); err != nil {
            log.Fatalln(err)
        }
    }()

    partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetNewest)
    if err != nil {
        panic(err)
    }

    defer func() {
        if err := partitionConsumer.Close(); err != nil {
            log.Fatalln(err)
        }
    }()

    // Trap SIGINT to trigger a shutdown.
    signals := make(chan os.Signal, 1)
    signal.Notify(signals, os.Interrupt)

    consumed := 0
ConsumerLoop:
    for {
        select {
        case msg := <-partitionConsumer.Messages():
            log.Printf("Consumed message offset %d,key:%s,value:%s\n", msg.Offset,msg.Key,msg.Value)
            consumed++
        case <-signals:
            break ConsumerLoop
        }
    }

    log.Printf("Consumed: %d\n", consumed)
}